package com.songtao.rabbitmq.confirm;

import java.io.IOException;
import java.util.Collections;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.songtao.rabbitmq.config.RabbitConfig;
import com.songtao.rabbitmq.util.RabbitConnectionUtils;

public class Send2 {
    
    
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        
        Connection connection = RabbitConnectionUtils.getConnection();
        
        Channel channel = connection.createChannel();
        
        //声明队列
        channel.queueDeclare(RabbitConfig.QUEUE_CONFIRM_1, false, false, false, null);
        
        
        final SortedSet<Long> comfirmSet = Collections.synchronizedSortedSet(new TreeSet<Long>());
        
        channel.addConfirmListener(new ConfirmListener() {
            
            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                if(multiple){
                    comfirmSet.headSet(deliveryTag + 1).clear();
                    System.out.println("成功移除multiple");
                }else{
                    comfirmSet.remove(deliveryTag);
                    System.out.println("成功移除");
                }
                
            }
            
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                if(multiple){
                    comfirmSet.headSet(deliveryTag + 1).clear();
                    System.out.println("失败移除multiple");
                }else{
                    comfirmSet.remove(deliveryTag);
                    System.out.println("失败移除");
                }
                
            }
        });
        
        
        //消息
        String msg = "hello rabbitmq tx";
       
        while(true){
            long seqNo = channel.getNextPublishSeqNo();
            channel.basicPublish("", RabbitConfig.QUEUE_CONFIRM_1, null, msg.getBytes());
            comfirmSet.add(seqNo);
        }
        
      
        
    }
}
